package com.danbay.framework.kafka.config;

import com.danbay.framework.kafka.AccessPackageDeserializer;
import com.danbay.framework.kafka.coap.CoAPJsonDataDecoder;
import com.danbay.framework.kafka.coap.CoAPJsonDataEncoder;
import com.danbay.framework.protocol.command.CommandBuilder;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.Map;

/**
 * @author SJG
 *         2017/11/15.
 */
@AutoConfigureAfter(KafkaConfiguration.class)
@EnableConfigurationProperties(KafkaProperties.class)
@ConfigurationProperties
@ImportAutoConfiguration(KafkaProducerConfig.class)
public class KafkaConsumerConfig {
    final KafkaProperties kafkaProperties;

    @Autowired
    public KafkaConsumerConfig(KafkaProperties kafkaProperties) {
        this.kafkaProperties = kafkaProperties;
    }

    @Bean
    ConcurrentKafkaListenerContainerFactory<Integer, String>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        Map<String, Object> props = kafkaProperties.buildConsumerProperties();
        return new DefaultKafkaConsumerFactory(props, new StringDeserializer(), new AccessPackageDeserializer());
    }

    @ConditionalOnMissingBean
    @Bean
    public CommandBuilder getCommandBuilder() {
        return new CommandBuilder();
    }

    @ConditionalOnMissingBean
    @Bean
    public AutoKafkaListener getAutoKafkaListener() {
        return new AutoKafkaListener(new CoAPJsonDataDecoder(), new CoAPJsonDataEncoder());
    }
}
